package com.imooc.smartcabinet.service;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.hawtdispatch.Dispatch;
import org.fusesource.mqtt.client.*;

import java.net.URISyntaxException;

/****
 * 参考博客说明:https://gitee.com/iots/mqtt-client
 * https://www.cnblogs.com/yanyu01/p/9557328.html
 */
public class MqttClient {

    public static void main(String[] args) throws Exception {
// MQTT设置说明
        MQTT mqtt = new MQTT();

        // 设置主机号
        mqtt.setHost("devops.robustel.cn:2238");
        // 用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。此ID应少于23个字符，默认根据本机地址、端口和时间自动生成
        mqtt.setClientId("DeviceId:876543210");
        // 若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true
        mqtt.setCleanSession(false);
        // 定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待
        mqtt.setKeepAlive((short) 60);
        // 服务器认证用户名
        mqtt.setUserName("admin");
        // 服务器认证密码
        mqtt.setPassword("password");
        // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
        mqtt.setWillTopic("willTopic");
        // 设置“遗嘱”消息的内容，默认是长度为零的消息
        mqtt.setWillMessage("willMessage--这是遗嘱消息");
        // 设置“遗嘱”消息的QoS(消息质量等级)，默认为QoS.ATMOSTONCE 最多一次；AT_MOST_ONCE 最多一次,AT_LEAST_ONCE 至少一次,EXACTLY_ONCE 只有一次
        mqtt.setWillQos(QoS.AT_LEAST_ONCE);
        // 若想要在发布“遗嘱”消息时拥有retain选项，则为true
        mqtt.setWillRetain(true);
        // 设置版本
        mqtt.setVersion("3.1.1");
        // 失败重连接设置说明
        // 客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
        mqtt.setConnectAttemptsMax(10L);
        // 客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
        mqtt.setReconnectAttemptsMax(3L);
        // 首次重连接间隔毫秒数，默认为10ms
        mqtt.setReconnectDelay(10L);
        // 重连接间隔毫秒数，默认为30000ms
        mqtt.setReconnectDelayMax(30000L);
        // 设置重连接指数回归。设置为1则停用指数回归，默认为2
        mqtt.setReconnectBackOffMultiplier(2);

        // Socket设置说明
        // 设置socket接收缓冲区大小，默认为65536（64k）
        mqtt.setReceiveBufferSize(65536);
        // 设置socket发送缓冲区大小，默认为65536（64k）
        mqtt.setSendBufferSize(65536);
        // 设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输
        mqtt.setTrafficClass(8);

        // 带宽限制设置说明
        // 设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制
        mqtt.setMaxReadRate(0);
        // 设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制
        mqtt.setMaxWriteRate(0);

        // 选择消息分发队列
        // 若没有调用方法setDispatchQueue，客户端将为连接新建一个队列。如果想实现多个连接使用公用的队列，显式地指定队列是一个非常方便的实现方法
        mqtt.setDispatchQueue(Dispatch.createQueue("foo"));

        msgListener(mqtt);
        pushMsg(mqtt,"This's Mqtt Message.");
    }

    /**
     * fusesource提供三种mqtt client api，分别为阻塞API，基于Futur的API和回调API。
     * 第一个阻塞API。代码如下：
     */
    private void aVoid(MQTT mqtt) throws Exception {
        // 使用future连接
        FutureConnection connection = mqtt.futureConnection();
        Future<Void> f1 = connection.connect();
        f1.await();
        // 订阅消息
        Future<byte[]> f2 = connection.subscribe(new Topic[]{new Topic("主题", QoS.AT_LEAST_ONCE)});
        //
        byte[] qoses = f2.await();

        // 发送身份验证消息.
        // Future<Void> f3 = connection.publish("foo", "Hello".getBytes(),
        // QoS.AT_LEAST_ONCE, false);
        // 接收订阅消息..
        Future<Message> receive = connection.receive();
        // 打印消息.
        Message message = receive.await();
        System.out.println(String.valueOf(message.getPayloadBuffer()));
        // 回应
        message.ack();
        //
        Future<Void> f4 = connection.disconnect();
        f4.await();
    }

    /***
     * Future  Connection
     */
    public void futureConnection(MQTT mqtt) throws Exception {
// 使用future连接
        FutureConnection connection = mqtt.futureConnection();
        Future<Void> f1 = connection.connect();
        f1.await();
        // 订阅消息
        Future<byte[]> f2 = connection.subscribe(new Topic[]{new Topic("主题", QoS.AT_LEAST_ONCE)});
        //
        byte[] qoses = f2.await();

        // 发送身份验证消息.
        // Future<Void> f3 = connection.publish("foo", "Hello".getBytes(),
        // QoS.AT_LEAST_ONCE, false);
        // 接收订阅消息..
        Future<Message> receive = connection.receive();
        // 打印消息.
        Message message = receive.await();
        System.out.println(String.valueOf(message.getPayloadBuffer()));
        // 回应
        message.ack();
        //
        Future<Void> f4 = connection.disconnect();
        f4.await();
    }

    public static void msgListener(final MQTT mqtt) throws Exception {
        // 使用future连接
        CallbackConnection connection = mqtt.callbackConnection();
        // 监听
        connection.listener(new Listener() {
            @Override
            public void onPublish(UTF8Buffer topicmsg, Buffer msg, Runnable ack) {
                // utf-8 is used for dealing with the garbled
                String topic = topicmsg.utf8().toString();
                String payload = msg.utf8().toString();
                System.out.println(topic + "  " + payload);
                pushMsg(mqtt, payload);
//                String Amsg = AuthenticationSendDemo.Authentication(topic, payload);
                // 表示监听成功
                ack.run();
            }

            @Override
            public void onFailure(Throwable value) {
                // 表示监听失败
                System.err.println("表示监听失败!");
            }

            // execute only once when connection is ended
            @Override
            public void onDisconnected() {
                // 表示监听到断开连接
                System.out.println("断开连接！！");
            }

            // execute only once when connecting started
            @Override
            public void onConnected() {
                // 表示监听到连接成功
                System.out.println("haha");
            }
        });
    }

    /***
     * 消息推送工具，待返回
     * @param mqtt
     * @param content
     */
    public static void pushMsg(MQTT mqtt, String content) {
        CallbackConnection connection = mqtt.callbackConnection();
        connection.getDispatchQueue().execute(() -> {
            connection.publish("/topic", content.getBytes(), QoS.AT_MOST_ONCE, false, new Callback<Void>() {
                @Override
                public void onSuccess(Void args) {
                    // 表示发布主题成功
                    System.out.println("发布成功！");
                    System.out.println("发布的消息:" + content);
                }

                @Override
                public void onFailure(Throwable throwable) {
                    // 表示发布主题失败
                    System.out.println("发布失败！");
                }
            });
        });
    }
}
